package library.net.retrofit.response;

import android.support.annotation.NonNull;

import java.io.IOException;
import java.lang.ref.SoftReference;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import library.net.retrofit.RxSchedulers;
import library.net.retrofit.callback.IRxCallback;
import library.net.retrofit.entity.FileEntity;
import okhttp3.MediaType;
import okhttp3.ResponseBody;
import okio.Buffer;
import okio.BufferedSource;
import okio.ForwardingSource;
import okio.Okio;
import okio.Source;

/**
 * 自定义进度的body
 *
 * @author Created by WZG on 2016/10/20.
 * @version 1.0
 */
public class RxProgressResponseBody extends ResponseBody {

    /**
     * real response body
     */
    private ResponseBody responseBody;
    /**
     * call back interface
     */
    private SoftReference<IRxCallback> mObserverListener;
    /**
     * BufferedSource
     */
    private BufferedSource bufferedSource;
    /**
     * frequency for progress listener
     */
    private float frequency = 0.5f;

    private FileEntity fileEntity;

    public RxProgressResponseBody(ResponseBody responseBody, @NonNull FileEntity fileEntity, IRxCallback progressListener) {
        this.responseBody = responseBody;
        this.fileEntity = fileEntity;
        /* get size of total file */
        if (fileEntity.totalBytes <= 0) {
            // 文件尚未下载，取contentLength作为整个文件长度
            fileEntity.totalBytes = contentLength();
            fileEntity.commitToSharedPre(fileEntity.url);
        }
        this.mObserverListener = new SoftReference<>(progressListener);
    }

    /**
     * 重写调用实际的响应体的contentType
     *
     * @return MediaType
     */
    @Override
    public MediaType contentType() {
        return null == responseBody ? null : responseBody.contentType();
    }

    /**
     * 重写调用实际的响应体的contentLength
     *
     * @return contentLength
     */
    @Override
    public long contentLength() {
        return null == responseBody ? 0 : responseBody.contentLength();
    }

    @Override
    public BufferedSource source() {
        if (bufferedSource == null && null != responseBody) {
            bufferedSource = Okio.buffer(source(responseBody.source()));
        }
        return bufferedSource;
    }

    private Source source(Source source) {
        final long previousTime = System.currentTimeMillis();
        return new ForwardingSource(source) {
            /* count for updating the progress */
            int updateCount = 0;

            /* 本次下载前已经下载的长度 */
            long cachedCompletedSize = fileEntity.completedBytes;

            @Override
            public long read(Buffer sink, long byteCount) throws IOException {

                long bytesRead = super.read(sink, byteCount);
                /* read() returns the number of bytes read, or -1 if this source is exhausted. */
                fileEntity.completedBytes += bytesRead != -1 ? bytesRead : 0;

                if (fileEntity.totalBytes > 1024 * 5) {
                    /* if the total size > 5M , frequency for updating progress is at least 1% */
                    frequency = 1;
                } else {
                    /* if the total size < 5M , frequency for updating progress is at least 0.5% */
                    frequency = (long) 0.5;
                }
                /* time */
                long totalTime = (System.currentTimeMillis() - previousTime) / 1000;
                if (totalTime == 0) {
                    totalTime += 1;
                }
                int progress = (int) (fileEntity.completedBytes * 100 / fileEntity.totalBytes);
                /* Speed: bytes/s */
                fileEntity.networkSpeed = (fileEntity.completedBytes - cachedCompletedSize) / totalTime;
                /* update the progress at least 1% */
                if (updateCount == 0 || progress > updateCount) {
                    updateCount += frequency;
                    Observable.just(fileEntity)
                            .compose(RxSchedulers.<FileEntity>io_main())
                            .subscribe(new Observer<FileEntity>() {
                                @Override
                                public void onSubscribe(Disposable d) {
                                }

                                @Override
                                public void onNext(FileEntity fileEntity) {
                                    if (mObserverListener.get() != null) {
                                        mObserverListener.get().onProgress(
                                                null,
                                                fileEntity.networkSpeed,
                                                fileEntity.completedBytes, fileEntity.totalBytes);
                                    }
                                }

                                @Override
                                public void onError(Throwable e) {
                                }

                                @Override
                                public void onComplete() {
                                }
                            });
                }
                return bytesRead;
            }
        };
    }
}
